package com.mytijian.task.task;

import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import com.mytijian.base.exception.BizException;
import com.mytijian.task.enums.OperateTypeEnum;
import com.mytijian.task.enums.TaskStatusEnum;
import com.mytijian.task.exception.MtjkTaskProcessException;
import com.mytijian.task.model.SubTask;
import com.mytijian.task.repository.model.GroupExecutorDTO;
import com.mytijian.task.repository.redis.RuningTaskRedisRepository;
import com.mytijian.task.task.step.TaskStep;
import com.mytijian.task.task.step.TaskStepContainer;
import com.mytijian.task.util.IpAddressUtil;
import com.mytijian.task.model.TaskValue;
import com.mytijian.task.enums.TaskTypeEnum;
import com.mytijian.task.executor.GroupExecutorService;
import com.mytijian.task.lifecycle.ApplicationLifycycle;
import com.mytijian.task.model.Task;
import com.mytijian.task.repository.RuningTaskRepository;
import com.mytijian.task.repository.TaskRepository;
import com.mytijian.task.task.process.DefaultProcessLifecycle;
import com.mytijian.task.task.process.ProcessLifecycle;
import com.mytijian.task.model.TaskIdentify;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;

import java.lang.reflect.ParameterizedType;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;

public class TaskIdentifyOperateServiceImpl extends ApplicationLifycycle implements TaskIdentifyOperateService {
    private TaskRepository taskRepository;
    private List<AbstractTaskProcess> taskProcesses;
    private RuningTaskRepository runingTaskRepository;
    private Map<String, Map<TaskIdentify, Future>> groupKeyAndFutureMap = new ConcurrentHashMap<>();
    private Logger logger = LoggerFactory.getLogger(this.getClass());
    /**
     * 用于暂停，停止任务。
     */
    private Map<TaskIdentify, TaskStatusEnum> localStatusMap = new ConcurrentHashMap<>();
    /**
     * 任务操作锁，同一时间一个任务只能被操作一次。
     */
    private Map<Integer, OperateTypeEnum> operateLock = new ConcurrentHashMap<>();
    public TaskIdentifyOperateServiceImpl(TaskRepository taskRepository, List<AbstractTaskProcess> abstractTaskLaunches, RuningTaskRedisRepository runingTaskRedisRepository) {
        this.taskRepository = taskRepository;
        this.taskProcesses = abstractTaskLaunches;
        this.runingTaskRepository = runingTaskRedisRepository;
    }
    @Override
    public void init() {
        List<GroupExecutorDTO> list = taskRepository.getGroupExecutorsAll();
        List<GroupExecutorService.GroupExecutorParam> groupExecutorParams = new ArrayList<>(list.size());
        for (GroupExecutorDTO groupExecutorDTO : list) {
            groupExecutorParams.add(new GroupExecutorService.GroupExecutorParam(groupExecutorDTO.getWorkThreads(), groupExecutorDTO.getCapacity(), groupExecutorDTO.getGroupKey()));
            groupKeyAndFutureMap.put(groupExecutorDTO.getGroupKey(), new ConcurrentHashMap<>());
        }
        GroupExecutorService.init(groupExecutorParams);
    }
    @Override
    public int initSort() {
        return 1;
    }

    @Override
    public synchronized boolean submit(TaskIdentify taskIdentify, TaskStatusEnum taskStatusEnum) {
        Integer workingThreads = groupKeyAndFutureMap.get(taskIdentify.getGroupKey()).size();
        long maxThreads = GroupExecutorService.getCapacity(taskIdentify.getGroupKey());
        if (workingThreads != 0 && workingThreads >= maxThreads) {
            logger.error("提交任务失败 当前没有空闲线程 taskIdentify = {}, ip = {} workingSize = {}, maxThreads = {}",taskIdentify.getTaskKey(), IpAddressUtil.getLocalHost(), workingThreads, maxThreads);
            return false;
        }
        if (isShutDown(taskIdentify.getGroupKey())) {
            logger.error("线程已关闭，不接受新任务 taskIdentify = {}", taskIdentify.getTaskKey());
            return false;
        }
        localStatusMap.put(taskIdentify, taskStatusEnum);
        process(taskIdentify);
        return true;
        //
    }
    @Override
    public void stop(Integer taskId) {
        OperateTypeEnum operateTypeEnum = operateLock.get(taskId);
        if (operateTypeEnum != null && !OperateTypeEnum.STOP.equals(operateTypeEnum)) {
            throw new BizException("12", "等待其他操作：" + operateTypeEnum.getDescription());
        }
        List<TaskIdentify> breakTasks = new ArrayList<>();
        try {
            operateLock.put(taskId, OperateTypeEnum.STOP);
            Map<TaskIdentify, TaskValue> taskIdentifies = runingTaskRepository.getByIpAndTaskId(IpAddressUtil.getLocalHost(), taskId);
            if (CollectionUtils.isEmpty(taskIdentifies)) {
                throw new BizException("12", "任务不存在");
            }
            List<TaskStatusEnum> status = Lists.newArrayList(TaskStatusEnum.un_begin, TaskStatusEnum.processing, TaskStatusEnum.suspend);
            breakTasks = breakSubTask(taskIdentifies.keySet(), status);
            if (CollectionUtils.isEmpty(breakTasks)) {
                logger.warn("没有需要停止的任务 ip = {} taskId = {}", IpAddressUtil.getLocalHost(), taskId);

                return;
            }
            for (TaskIdentify taskIdentify : breakTasks) {
                localStatusMap.put(taskIdentify, TaskStatusEnum.stop);
            }
            wait(breakTasks, null);
        }catch (Exception e) {
            logger.error("stop error taskId = {}, e = {}", taskId, e);
            throw e;
        }finally {
            operateLock.remove(taskId);
            for (TaskIdentify taskIdentify : breakTasks) {
                groupKeyAndFutureMap.get(taskIdentify.getGroupKey()).remove(taskIdentify);
                TaskValue taskValue = runingTaskRepository.getByIdentify(IpAddressUtil.getLocalHost(), taskIdentify);
                // 解决暂停后，点击停止 redis中状态没有变化
                if (!TaskStatusEnum.getEndStatus().contains(TaskStatusEnum.getByCode(taskValue.getTaskStuts())) ) {
                    taskValue.setTaskStuts(TaskStatusEnum.stop.getCode());
                    runingTaskRepository.save(taskIdentify, taskValue, IpAddressUtil.getLocalHost());
                }
                localStatusMap.remove(taskIdentify);
            }
        }

    }

    public static void main(String[] args) {
        System.out.println(TaskStatusEnum.getEndStatus().contains(TaskStatusEnum.stop));
    }
    private synchronized List<TaskIdentify> breakSubTask(Collection<TaskIdentify> taskIdentifies, List<TaskStatusEnum> allowBreakStatus) {
        List<TaskIdentify> breakTasks = new ArrayList<>();
        for (TaskIdentify taskIdentify : taskIdentifies) {
            TaskStatusEnum statusEnum = localStatusMap.get(taskIdentify);
            if (allowBreakStatus.contains(statusEnum)) {
                breakTasks.add(taskIdentify);
            }
        }
        return breakTasks;
    }

    private synchronized List<TaskIdentify> beginSubTask(Collection<TaskIdentify> taskIdentifies) {
        List<TaskStatusEnum> allowBeginStatus = Lists.newArrayList(TaskStatusEnum.un_begin, TaskStatusEnum.suspend, TaskStatusEnum.processing);
        List<TaskIdentify> allowBegins = new ArrayList<>();
        for (TaskIdentify taskIdentify : taskIdentifies) {
            TaskStatusEnum taskStatusEnum = localStatusMap.get(taskIdentify);
            Future future = groupKeyAndFutureMap.get(taskIdentify.getGroupKey()).get(taskIdentify);
            if (allowBeginStatus.contains(taskStatusEnum) && (future  != null && future.isDone())) {
                allowBegins.add(taskIdentify);
            }
        }
        return allowBegins;
    }

    @Override
    public void suspend(Integer taskId) {
        OperateTypeEnum operateTypeEnum = operateLock.get(taskId);
        if (operateTypeEnum != null && !OperateTypeEnum.SUSPEND.equals(operateTypeEnum)) {
            throw new BizException("12", "等待其他操作：" + operateTypeEnum.getDescription());
        }
        try {
            operateLock.put(taskId, OperateTypeEnum.SUSPEND);
            Set<TaskIdentify> taskIdentifies = runingTaskRepository.getByIpAndTaskId(IpAddressUtil.getLocalHost(), taskId).keySet();
            if (CollectionUtils.isEmpty(taskIdentifies)) {
                throw new BizException("12", "任务不存在");
            }
            List<TaskStatusEnum> status = Lists.newArrayList(TaskStatusEnum.un_begin, TaskStatusEnum.processing);
            List<TaskIdentify> suspendTasks = breakSubTask(taskIdentifies, status);
            if (CollectionUtils.isEmpty(suspendTasks)) {
                logger.warn("没有需要暂停的任务 ip = {} taskId = {}", IpAddressUtil.getLocalHost(), taskId);
                return;
            }
            for (TaskIdentify taskIdentify : suspendTasks) {
                localStatusMap.put(taskIdentify, TaskStatusEnum.suspend);
            }
            wait(suspendTasks, null);
        }catch (Exception e) {
            logger.error("suspend error taskId = {}, e = {}", e);
        }finally {
            operateLock.remove(taskId);
        }


    }

    private void wait(Collection<TaskIdentify> taskIdentifies, Long waitTimeMillis) {
        long first = System.currentTimeMillis();
        List<Integer> doneList = new ArrayList<>();
        while (true) {
            // 等待当前任务执行完
            for (TaskIdentify taskIdentify : taskIdentifies) {
                Future future = groupKeyAndFutureMap.get(taskIdentify.getGroupKey()).get(taskIdentify);
                if (future.isDone()) {
                    doneList.add(1);
                }
            }
            try {
                Thread.sleep(50);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            if (System.currentTimeMillis() - first > 1000 * 2) {
                logger.error("等待当前任务执行完超过2s, 当前等待时间 wait = {}s ", (System.currentTimeMillis() - first) / 1000);
            }
            if (doneList.size() == taskIdentifies.size()) {
                break;
            }
            if (waitTimeMillis != null) {
                if (System.currentTimeMillis() - first > waitTimeMillis) {
                    break;
                }
            }
            doneList.clear();
        }
    }

    private AtomicInteger atomicInteger = new AtomicInteger(0);
    @Override
    public void destory() {

        logger.warn("程序关闭 {}", atomicInteger.getAndIncrement());
        GroupExecutorService.shutdownAll();
        List<TaskStatusEnum> status = Lists.newArrayList(TaskStatusEnum.un_begin, TaskStatusEnum.processing);
        List<TaskIdentify> taskIdentifies = new ArrayList<>(1024);
        localStatusMap.forEach(((taskIdentify, taskStatusEnum) -> {
            if (status.contains(taskStatusEnum)) {
                localStatusMap.put(taskIdentify, TaskStatusEnum.suspend);
                taskIdentifies.add(taskIdentify);
            }
        }));
        // 等待任务执行完
        wait(taskIdentifies, null);
        logger.warn("资源清理完毕");
    }
    @Override
    public int destorySort() {
        return 2;
    }

    /**
     * @param taskId
     */
    @Override
    public void process(Integer taskId) {
        OperateTypeEnum operateTypeEnum = operateLock.get(taskId);
        if (operateTypeEnum != null && !OperateTypeEnum.PROCESS.equals(operateTypeEnum)) {
            throw new BizException("12", "等待其他操作：" + operateTypeEnum.getDescription());
        }
        try {
            operateLock.put(taskId, OperateTypeEnum.PROCESS);
            Map<TaskIdentify, TaskValue> taskValueMap = runingTaskRepository.getByIpAndTaskId(IpAddressUtil.getLocalHost(), taskId);
            List<TaskIdentify> taskIdentifies = beginSubTask(taskValueMap.keySet());
            if (CollectionUtils.isEmpty(taskIdentifies)) {
                logger.warn("没有需要开始的任务 ip = {} taskId = {}", IpAddressUtil.getLocalHost(), taskId);
                return;
            }
            for (TaskIdentify taskIdentify : taskIdentifies) {
                if (isShutDown(taskIdentify.getGroupKey())) {
                    return;
                }
                TaskValue taskValue = taskValueMap.get(taskIdentify);
                taskValue.setTaskStuts(TaskStatusEnum.processing.getCode());
                runingTaskRepository.save(taskIdentify, taskValue, IpAddressUtil.getLocalHost());
                localStatusMap.put(taskIdentify, TaskStatusEnum.processing);
                process(taskIdentify);
            }
        }catch (Exception e) {
            logger.error("process error taskId = {}, e = {}", taskId, e);
            throw e;
        }finally {
            operateLock.remove(taskId);
        }


    }

    @Override
    public void clear(List<TaskIdentify> taskIdentifies) {
        for (TaskIdentify taskIdentify : taskIdentifies) {
            groupKeyAndFutureMap.get(taskIdentify.getGroupKey()).remove(taskIdentify);
            localStatusMap.remove(taskIdentify);
        }
    }
    @Override
    public boolean isShutDown(String groupKey) {
        return GroupExecutorService.isShutdown(groupKey);
    }




    private boolean process(TaskIdentify taskIdentify) {
        if (groupKeyAndFutureMap.get(taskIdentify.getGroupKey()).get(taskIdentify) != null && !groupKeyAndFutureMap.get(taskIdentify.getGroupKey()).get(taskIdentify).isDone()) {
            logger.warn("任务taskIdentify = {},已经在执行", taskIdentify.getTaskKey());
            return true;
        }
        Future future = GroupExecutorService.submit(taskIdentify.getGroupKey(), new Runnable() {
            @Override
            public void run() {
                long first = System.currentTimeMillis();
                ProcessLifecycle processLifecycle = new DefaultProcessLifecycle(runingTaskRepository, taskRepository);
                List<SubTask> subTasks = null;
                TaskValue taskValue = null;
                List<TaskStatusEnum> status = TaskStatusEnum.getUnProcessStatus();
                Task task = null;
                try {
                    TaskStatusEnum taskStatus = null;
                    task = taskRepository.getTasksByIds(Lists.newArrayList(taskIdentify.getTaskId())).get(0);
                    if (status.contains(TaskStatusEnum.getByCode(task.getStatus()))) {
                        taskStatus = TaskStatusEnum.getByCode(task.getStatus());
                    }
                    subTasks = taskRepository.getSubTaskInId(taskIdentify.getTaskId(), taskIdentify.getBeginSubTaskId(), taskIdentify.getEndSubTaskId());
                    taskValue = runingTaskRepository.getByIdentify(IpAddressUtil.getLocalHost(), taskIdentify);
                    if (taskValue == null) {
                        return;
                    }
                    if (status.contains(TaskStatusEnum.getByCode(taskValue.getTaskStuts()))) {
                        taskStatus = TaskStatusEnum.getByCode(taskValue.getTaskStuts());
                    }
                    if (localStatusMap.get(taskIdentify) != null && status.contains(localStatusMap.get(taskIdentify))) {
                        taskStatus = localStatusMap.get(taskIdentify);
                    }
                    if (taskStatus != null) {
                        processLifecycle.taskProcessComplete(taskIdentify, taskStatus, null);
                        localStatusMap.put(taskIdentify, taskStatus);
                        return;
                    }
                } catch (Exception e) {
                    logger.error("任务执行校验出错 taskIdentify = {} e = {}",taskIdentify.getTaskKey(), e);
                    return;
                }

                TaskStatusEnum taskStatus = TaskStatusEnum.complete;
                Exception exception = null;
                try {
                    processLifecycle.beforeTaskProcess(taskIdentify);
                    Integer completeSubTaskId = taskValue.getCompleteSubTaskId();
                    Integer processingSubTaskId = taskValue.getProcessingSubTaskId();
                    List<SubTask> needProcessSubTasks = needProcess(subTasks, taskValue);
                    for (SubTask subTask : needProcessSubTasks) {
                        // 任务终止
                        if (status.contains(localStatusMap.get(taskIdentify))) {
                            logger.info("任务暂停 taskIdentify = {}", taskIdentify.getTaskKey());
                            taskStatus = localStatusMap.get(taskIdentify);
                            break;
                        }
                        processLifecycle.beforeSubTaskProcess(subTask, taskIdentify);
                        Exception temp = null;
                        Integer currentRetryTimes = 0;
                        while (true) {
                            try {
                                Object value = TaskStepContainer.getProcessValue(task.getTaskTypeEnum(), subTask.getParam());
                                if (!Objects.equals(completeSubTaskId, processingSubTaskId)) {
                                    if (TaskStepContainer.needProcess(task.getTaskTypeEnum(), value)) {
                                        TaskStepContainer.process(task.getTaskTypeEnum(), value);
                                    }
                                } else {
                                    TaskStepContainer.process(task.getTaskTypeEnum(), value);
                                }

                            } catch (Exception e) {
                                logger.warn("处理任务出错 subTask = {} , e ={}", JSONObject.toJSONString(subTask), e);
                                temp = e;
                                if (e instanceof MtjkTaskProcessException) {
                                    localStatusMap.put(taskIdentify, TaskStatusEnum.suspend);
                                }
                            }
                            if (!TaskStepContainer.needRetry(task.getTaskTypeEnum(), currentRetryTimes, temp)) {
                                break;
                            }
                            currentRetryTimes++;
                        }
                        processLifecycle.afterSubTaskProcess(subTask, taskIdentify, temp);
                    }
                } catch (Exception e) {
                    logger.error("处理任务出错 taskIdentify = {} e = {}", taskIdentify.getTaskKey(), e);
                    taskStatus = TaskStatusEnum.processing;
                    exception = e;
                } finally {
                    processLifecycle.taskProcessComplete(taskIdentify, taskStatus, exception);
                    localStatusMap.put(taskIdentify, taskStatus);
                    logger.info("taskIdentify = {} taskStatus = {} , 处理结束，处理耗时 {}", taskIdentify.getTaskKey(), taskStatus, System.currentTimeMillis() - first);
                }
            }
        });
        groupKeyAndFutureMap.get(taskIdentify.getGroupKey()).put(taskIdentify, future);
        return true;
    }

    private List<SubTask> needProcess(List<SubTask> subTasks, TaskValue taskValue) {
        if (taskValue.getProcessingSubTaskId() == null) {
            return subTasks;
        }
        List<SubTask> needProcessSubTasks = new ArrayList<>(subTasks.size());

        if (taskValue.getCompleteSubTaskId() != null) {
            for (SubTask subTask : subTasks) {
                if (subTask.getId() > taskValue.getCompleteSubTaskId()) {
                    needProcessSubTasks.add(subTask);
                }
            }
            return needProcessSubTasks;
        }
        for (SubTask subTask : subTasks) {
            if (subTask.getId() >= taskValue.getProcessingSubTaskId()) {
                needProcessSubTasks.add(subTask);
            }
        }
        return needProcessSubTasks;
    }

}
